package cn.ipanel.bigdata.job

import cn.ipanel.bigdata.boot.date.{Day, Month, Quarter, Time, Week, Year}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.boot.period.Period
import cn.ipanel.bigdata.dw.dwd._
import cn.ipanel.bigdata.dw.dwm.AbsDwmHive
import cn.ipanel.bigdata.job.dwd.AbsDwdJob
import org.apache.spark.sql.{DataFrame}
import org.apache.spark.sql.functions.{col, last, lit, max, min, month, quarter, sum, when}
import org.apache.spark.storage.StorageLevel

/**
 * Author: lzz
 * Date: 2022/3/29 8:50
 */
package object dwm {

  abstract class AbsDwmJob extends AbsDwdJob {

    private[this] final lazy val _CACHE: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER

    /**
     *
     *  @return
     */
    override def optSourceTable: AbsDwdHive
    override lazy val _S: AbsDwdHive = optSourceTable

    override def optTargetTable: AbsDwmHive
    override lazy val _T: AbsDwmHive = optTargetTable

    // 计算周期，目前只有 天
    private[this] lazy val _PERIOD: Seq[Period] = {
//      getTemplate.asPeriod ++ Seq(Period.DAY, Period.WEEK, Period.MONTH, Period.QUARTER, Period.YEAR)
      getTemplate.asPeriod ++ Seq(Period.DAY)
    }

    // 固定观测视角，目前只有 区域、产商、牌照方 （目前配置文件固定视角为空）
    private[this] lazy val _FIXED: Seq[String] = {
      getTemplate.asFixed ++
        Seq(_T.F_REGION)
    }

    /**
     * 可选的各实现类务定制的观测视角
     * @return
     */
    def optExtraAngle: Seq[String] = Seq()

    /**
     * 可选的字段聚合方式，每一个实现类扩展的字段，必须都是可以累加的
     * @return
     */
    def optAggColumnNames: Seq[String] = Seq()

    // 所有观测视角 = 外部配的观测视角（不包括外部配置固定翻倍视角）
    //              + 各实现类定制观测视角
    //              + 固定观测视角（包括外部配置固定翻倍视角 + 内置必须固定观测的视角）
    //              + Seq(设备唯一编号 + 统计周期类型 + 统计周期值）
    private[this] lazy val _ANGLE: Seq[String] = {
      getTemplate.asAngle ++
        optExtraAngle ++
        _FIXED ++
        Seq(_T.F_DEVICE_ID, _T.F_PERIOD_TYPE, _T.F_DATE)
    }

    // 字段聚合方式 除了观测视角（配置文件的观测视角 + 各实现类务定制的观测视角 + ）之外的字段，统一使用 last 的方式
    private[this] lazy val _AGG_COLUMNS: Seq[org.apache.spark.sql.Column] = {
      var cols: Seq[org.apache.spark.sql.Column] = Seq()
      _T.getTBColumns.foreach(colName => {
        if (!_ANGLE.contains(colName) &&
            !_T.F_COUNT.equals(colName) && !optAggColumnNames.contains(colName) &&
            !_T.F_CREATE_TIME.equals(colName) && !_T.F_MODIFY_TIME.equals(colName))
          cols ++= Seq(last(colName).isNotNull as colName)
      })

      var aggColumns: Seq[org.apache.spark.sql.Column] = Seq()
      // 每一个实现类扩展的字段，必须都是可以累加的
      optAggColumnNames.foreach(x => aggColumns = aggColumns :+ (sum(x) as x))
      cols ++ aggColumns ++ Seq(sum(_T.F_COUNT) as _T.F_COUNT,
                  min(_T.F_CREATE_TIME) as _T.F_CREATE_TIME,
                  max(_T.F_MODIFY_TIME) as _T.F_MODIFY_TIME)
    }

    /**
     * 原始数据来源
     *  @return
     */
    override def source(): DataFrame = _S.find(_DAY)
    override def onPrepare(): Unit = _T.drop(_DAY, Period.DAY)
    override def onStartup(): Unit = {
      Logger.I("optAggColumn: " + optAggColumnNames)
      Logger.I("_AGG_COLUMNS: " + _AGG_COLUMNS)
      // 当天需要汇总的数据，转成汇总后的格式
      // 之所以要这样，因为天数据要汇总到更广周期(周、月、年等)中去
      var src = _JOIN.join(_INTENT.intent(source(), _DAY), _DAY)
                    .withColumn(_T.F_CREATE_TIME, lit(_DAY.toDate))
                    .withColumn(_T.F_MODIFY_TIME, lit(_DAY.toDate))
                    .withColumn(_T.F_PERIOD_TYPE, lit(Period.DAY.getType))
                    .withColumn(_T.F_DATE, lit(_DAY.toDate))
                    .selectExpr(_T.getTBColumns: _*)

      // 固定观测视角对应的数据翻倍，因为固定视角有ALL的情况
      // 山西现场不配置数据翻倍后存储逻辑，磁盘空间有限，本次数据存储后由其他任务内存翻倍计算好了
//      _FIXED.foreach((fixed: String) => {
//        src = src.union(src.withColumn(fixed, lit(_T.DEFAULT_ALL)))
//      })
      src.persist(_CACHE)
      src.show(100)
      _PERIOD.foreach((period: Period) => {
        if (period._type != Period.DAY._type){
          // 把天数据，组装到各个周期的数据中，后面一把算完
          src = src.union({
            _T.find(_DAY, period)
              .selectExpr(_T.getTBColumns: _*)
          })
        }
      })
      src.show(100)

      _T.save({
        src.groupBy(_ANGLE.head, _ANGLE.tail: _*)
           .agg(_AGG_COLUMNS.head, _AGG_COLUMNS.tail: _*)
      })
      src.unpersist()
    }

    /**
     * 重跑和回滚逻辑，暂时废弃，想了下各种情况都会发生，如果周月年的数据本来就不包含天数据或者其他异常，回滚没必要
     * 开发人员直接把周期内所有数据拿出来计算即可
     */
    @deprecated
    protected[this] def rollbackPeriodAndDropDay(): Unit = {
      // 天分区不为空，意味着重跑，得从周、月、年等分区中刨除本天的数据，主要是做减法，如次数、时长
      if (nonEmptyPartition(Period.DAY)) {
        val nowDate: Int = _DAY.toDate
        val week = if(nowDate - _DAY.asWeek.toDate > 1) _DAY.asWeek else null
        val month = if(nowDate - _DAY.asMonth.toDate > 1) _DAY.asMonth else null
        val quarter = if(nowDate - _DAY.asQuarter.toDate > 1) _DAY.asQuarter else null
        val year = if(nowDate - _DAY.asYear.toDate > 1) _DAY.asYear else null

        if (week == null && month == null && quarter == null && year == null) {
          Logger.I("本年第一天数据，无需重跑")
          return
        }

        val r_create_time = "r_create_time"
        val r_modify_time = "r_modify_time"

        // 回滚逻辑：
        // 1. 大周期 left join 天数据，使用 _ANGLE 字段作为连接条件（计算的时候，也是用的它分组，所以回滚也是）
        // 2. 不在 _ANGLE 的字段，就是计算结果字段，包括次数，时长等可以直接相减的字段；其中 f_create_time, f_modify_time 字段不管，会在下次计算时更新
        // 3. 回滚完，删除周期只有一天的周、月、年分区（因为如果当前周期只包含一天的数据，无需相减回滚，直接上坟去）
        val oldDf: DataFrame = find(week = week, month = month, quarter = quarter, year = year)
        var dayDf: DataFrame = find(day = _DAY)
          .withColumnRenamed(_T.F_CREATE_TIME, r_create_time)
          .withColumnRenamed(_T.F_MODIFY_TIME, r_modify_time)
          .drop(_T.F_PERIOD_TYPE, _T.F_DATE)

        // 所有 sum 计算的字段，包括了 f_count 和扩展字段
        val sumFields = Seq(_T.F_COUNT) ++ optAggColumnNames
        // 天其他扩展的计算字段，重命名
        optAggColumnNames.foreach(x => {
          dayDf = dayDf.withColumnRenamed(x, "r_" + x)
        })

        var resultDf = oldDf.join(dayDf, _ANGLE, joinType = "left")
        sumFields.foreach(x => {
          resultDf = resultDf
            .withColumn("r_" + x, when(col("r_" + x).isNull, lit(0)).otherwise(col("r_" + x)))
            .withColumn(x, col(x) - col("r_" + x))
          // 这里不能强制把小于0的转0，因为有的指标累加值可能就是0
//            .withColumn(x, when(col(x) < 0, lit(0)).otherwise(col(x)))
        })
        resultDf = resultDf.selectExpr(_T.getTBColumns: _*)
          // 存在这样的情况：重跑周期不是本周起第一天，但是本周起的数据起始天就是改天（第一次部署时），这样会导致存了一批数据是0的起始没必要的数据，
          // 只需要用 f_count = 0 即可过滤掉
          .filter(col(_T.F_COUNT) > lit(0))

        // 保存回滚后周/月/年数据
        _T.save(resultDf)

        // 重跑完，删除天数据
        _T.drop(_DAY, Period.DAY)

        // 删除不用回滚的周期数据
        if(week == null) {
          _T.drop(_DAY, Period.WEEK)
        }
        if(month == null) {
          _T.drop(_DAY, Period.MONTH)
        }
        if(quarter == null) {
          _T.drop(_DAY, Period.QUARTER)
        }
        if(year == null) {
          _T.drop(_DAY, Period.YEAR)
        }
      }
    }

    /**
     * 判断分区是否存在
     * @param period
     * @return
     */
    def nonEmptyPartition(period: Period): Boolean = {
      var df = spark.sql(s"show partitions ${_T.getDBName}.${_T.getTBName}")
      val part = period match {
        case d: Day     => s"${_T.F_PERIOD_TYPE}=${Period.DAY.getType}/${_T.F_DATE}=${d.toDate}"
        case w: Week    => s"${_T.F_PERIOD_TYPE}=${Period.WEEK.getType}/${_T.F_DATE}=${w.toDate}"
        case m: Month   => s"${_T.F_PERIOD_TYPE}=${Period.MONTH.getType}/${_T.F_DATE}=${m.toYearMonth}"
        case q: Quarter => s"${_T.F_PERIOD_TYPE}=${Period.QUARTER.getType}/${_T.F_DATE}=${q.toYearQuarter}"
        case y: Year    => s"${_T.F_PERIOD_TYPE}=${Period.YEAR.getType}/${_T.F_DATE}=${y.toYear}"
      }
      df = df.filter(x => x.getAs[String]("partition").contains(part))
      val exists = !df.isEmpty
      Logger.I(s"partition $part exists: $exists")
      df.show(false)

      exists
    }

    /**
     * 获取周、月、年等更长周期的汇总数据，用于回滚天数据
     * @param week
     * @param month
     * @param quarter
     * @param year
     * @return
     */
    private[this] def find(day: Day = null, week: Week = null, month: Month = null, quarter: Quarter = null,
                           year: Year = null)
    : DataFrame = {
      val sb = new StringBuilder
      var flag = false

      if (day != null) {
        if (flag) {
          sb.append(" or ")
        }
        sb.append(s"(${_T.F_PERIOD_TYPE} = ${Period.DAY.getType} and ${_T.F_DATE}=${day.toDate})")
        flag = true
      }
      if (week != null) {
        if (flag) {
          sb.append(" or ")
        }
        sb.append(s"(${_T.F_PERIOD_TYPE} = ${Period.WEEK.getType} and ${_T.F_DATE}=${week.toDate})")
        flag = true
      }
      if (month != null) {
        if (flag) {
          sb.append(" or ")
        }
        sb.append(s"(${_T.F_PERIOD_TYPE} = ${Period.MONTH.getType} and ${_T.F_DATE}=${month.toYearMonth})")
        flag = true
      }
      if (quarter != null) {
        if (flag) {
          sb.append(" or ")
        }
        sb.append(s"(${_T.F_PERIOD_TYPE} = ${Period.QUARTER.getType} and ${_T.F_DATE}=${quarter.toYearQuarter})")
        flag = true
      }
      if (year != null) {
        if (flag) {
          sb.append(" or ")
        }
        sb.append(s"(${_T.F_PERIOD_TYPE} = ${Period.YEAR.getType} and ${_T.F_DATE}==${year.toYear})")
        flag = true
      }

      Logger.I(s"load from ${_T.getDBName}.${_T.getTBName}, where ${sb.toString()}")
      _T.load.where(s"""
                        |${sb.toString()}
                        |""".stripMargin)
    }
  }

}